package com.project.base.hbase;

import com.google.common.collect.Lists;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.*;
import java.util.Map.Entry;

@Component
public class HBaseUtils {
    /**
     * @Title：HBaseUtils.java
     * @Author:zhoutianqi
     * @Description:
     * @Date：2017年2月10日上午10:23:18
     */
    private final Logger logger = Logger.getLogger(HBaseUtils.class);
    /*@Autowired  @Qualifier("jedisWriterTemplate") 
    private JedisTemplate jedisWriterTemplate;
    @Autowired  @Qualifier("jedisReaderTemplate") 
    private JedisTemplate jedisReaderTemplate;*/
    @Autowired
    private Connection connection;
    @Autowired
    private Configuration configuration;

    /**
     * 创建表
     *
     * @param tableName
     * @throws Exception
     */
    public void createTable(String tableName, String[] columns, int maxVersions) {
        Admin admin = null;
        try {
            admin = connection.getAdmin();
            TableName name = TableName.valueOf(tableName);
            if (admin.tableExists(name)) {
                logger.info("table: " + tableName + "is exist");
                return;
            }
            HTableDescriptor tableDesc = new HTableDescriptor(name);

            for (String column : columns) {
                HColumnDescriptor family = new HColumnDescriptor(column);
                family.setCompressionType(Compression.Algorithm.SNAPPY);
                family.setMaxVersions(maxVersions);
                tableDesc.addFamily(family);
            }
            admin.createTable(tableDesc);
        } catch (IOException e) {
            e.printStackTrace();
            logger.info("create table " + tableName + " failed");
        } finally {
            try {
                if (null != admin) admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 删除表
     *
     * @param tableName
     * @throws Exception
     */
    public void delTable(String tableName) {
        Admin admin = null;
        try {
            admin = connection.getAdmin();
            TableName tName = TableName.valueOf(tableName);
            admin.disableTable(tName);
            admin.deleteTable(tName);
        } catch (Exception e) {
            e.printStackTrace();
            logger.info("delete table " + tableName + " failed");
        } finally {
            try {
                if (null != admin) admin.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }

    }

    /**
     * 插入数据
     *
     * @throws IOException
     */
    public void insertAndUpdate(String tableName, List<Put> puts) {

        TableName tName = TableName.valueOf(tableName);
        Table table = null;
        try {
            table = connection.getTable(tName);
            table.put(puts);
        } catch (IOException e) {
            e.printStackTrace();
            logger.info("insert or update table: " + tableName + " failed");
        } finally {
            try {
                if (null != table) table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }


    }


    /**
     * 查询一行的数据
     *
     * @param tableName
     * @param row
     * @param family
     * @param maxVersions
     * @return
     */
    public Map<String, List<String>> queryByRowKey(String tableName, String row, String family, Integer maxVersions) {
        return queryByRowKey(tableName, row, family, maxVersions, new String[0]);
    }

    public List<Map<String, List<String>>> queryByRows(String tableName, List<String> rows, String family, Integer maxVersions) {
        List<Map<String, List<String>>> list = new ArrayList<Map<String, List<String>>>();
        for (String row : rows) {
            list.add(queryByRowKey(tableName, row, family, maxVersions));
        }
        return list;
    }

    public <T> T queryByRowKey(String tableName, String row, String family, Integer maxVersions, Class<T> clzss) throws Exception {
        Map<String, List<String>> userInfoMap = queryByRowKey(tableName, row, family, maxVersions);
        T instance = null;
        if (!MapUtils.isEmpty(userInfoMap)) {
            instance = clzss.newInstance();
            BeanUtils.populate(instance, userInfoMap);
        }
        return instance;
    }

    public <T> List<T> queryByRowKey(String tableName, String family, List<String> rowKey, Class<T> clzss) throws Exception {
        List<Map<String, String>> userInfoMap = get(tableName, family, rowKey, true);
        List<T> returnList = Lists.newArrayList();
        for (Map<String, String> dataMap : userInfoMap) {
            T instance = null;
            if (!MapUtils.isEmpty(dataMap)) {
                instance = clzss.newInstance();
                BeanUtils.populate(instance, dataMap);
                returnList.add(instance);
            }
        }
        return returnList;
    }

    private Map<String, String> noVersionResultToMap(Result result, String family, boolean includeRowKey) {
        Map<String, String> resultMap = new HashMap<String, String>();
        if (result.getRow() == null || StringUtils.isBlank(Bytes.toString(result.getRow()))) {
            return resultMap;
        }
        if (includeRowKey) {
            resultMap.put("rowKey", Bytes.toString(result.getRow()));
        }
        NavigableMap<byte[], NavigableMap<byte[], byte[]>> map = result.getNoVersionMap();
        if (CollectionUtils.isEmpty(map)) {
            return resultMap;
        }
        NavigableMap<byte[], byte[]> familyMap = map.get(Bytes.toBytes(family));
        Set<Entry<byte[], byte[]>> entrySet = familyMap.entrySet();

        for (Entry<byte[], byte[]> entry : entrySet) {
            byte[] key = entry.getKey();
            byte[] value = entry.getValue();
            resultMap.put(Bytes.toString(key), Bytes.toString(value));
        }
        return resultMap;
    }

    /**
     * 查询一行的数据
     *
     * @param tableName
     * @param row
     * @param family
     * @param maxVersions
     * @return
     */
    public Map<String, List<String>> queryByRowKey(String tableName, String row, String family, Integer maxVersions, String... columns) {
        if (StringUtils.isBlank(tableName) || StringUtils.isBlank(row) || StringUtils.isBlank(family)) {
            return null;
        }
        TableName tName = TableName.valueOf(tableName);
        Table table = null;
        try {
            table = connection.getTable(tName);
            Get get = new Get(Bytes.toBytes(row));
            get.addFamily(Bytes.toBytes(family));
            get.setMaxVersions(maxVersions);
            if (columns != null && columns.length != 0) {
                for (String column : columns) {
                    get.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
                }
            }
            Result result = table.get(get);
            return getResultMap(family, result);
        } catch (IOException e) {
            e.printStackTrace();
            logger.info("get value for table: " + tableName + " family: " + family + "failed");
            return null;
        } finally {
            close(table, null);
        }
    }

    private Map<String, List<String>> getResultMap(String family, Result result) {
        Map<String, List<String>> resultMap = new HashMap<String, List<String>>();
        //获取列簇,每一个columen的所有版本的值
        multiVersionResultToMap(family, result, resultMap);
        return resultMap;
    }

    private Map<String, List<String>> getMultiResultMap(String family, Result result, boolean includeRowKey) {
        Map<String, List<String>> resultMap = new HashMap<String, List<String>>();
        if (result.getRow() == null || StringUtils.isBlank(Bytes.toString(result.getRow()))) {
            return null;
        }
        if (includeRowKey) {
            resultMap.put("rowKey", Lists.newArrayList(Bytes.toString(result.getRow())));
        }
        //获取列簇,每一个columen的所有版本的值
        multiVersionResultToMap(family, result, resultMap);
        return resultMap;
    }

    private void multiVersionResultToMap(String family, Result result, Map<String, List<String>> resultMap) {
        NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> map = result.getMap();
        if (!CollectionUtils.isEmpty(map)) {
            NavigableMap<byte[], NavigableMap<Long, byte[]>> familyMap = map.get(Bytes.toBytes(family));
            Set<Entry<byte[], NavigableMap<Long, byte[]>>> entrySet = familyMap.entrySet();
            for (Entry<byte[], NavigableMap<Long, byte[]>> entry : entrySet) {
                List<String> valueList = new ArrayList<String>();
                byte[] key = entry.getKey();
                NavigableMap<Long, byte[]> value = entry.getValue();
                Set<Entry<Long, byte[]>> entrySet2 = value.entrySet();
                for (Entry<Long, byte[]> entry2 : entrySet2) {
                    byte[] value2 = entry2.getValue();
                    valueList.add(Bytes.toString(value2));
                }
                resultMap.put(Bytes.toString(key), valueList);
            }
        }
    }

    /**
     * 根据rowKey删除一条数据
     *
     * @param tableName
     * @param rowKey
     */
    public void deleteByRowkey(String tableName, String rowKey) {
        TableName tName = TableName.valueOf(tableName);
        Delete del = new Delete(Bytes.toBytes(rowKey));
        Table table = null;
        try {
            table = connection.getTable(tName);
            table.delete(del);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            close(table);
        }
    }

    /**
     * 根据指定的key集合和版本数查询数据
     *
     * @param tableName 表名
     * @param family    列簇
     * @param rowKeys   rowKey集合
     * @return List
     */
    public List<Map<String, String>> get(String tableName, String family, List<String> rowKeys) {
        if (StringUtils.isBlank(tableName) || rowKeys == null || rowKeys.isEmpty() || StringUtils.isBlank(family)) {
            return null;
        }

        TableName tName = TableName.valueOf(tableName);
        Table table = null;
        try {
            table = connection.getTable(tName);
            List<Get> gets = new ArrayList<Get>();
            for (String rowkey : rowKeys) {
                Get get = new Get(Bytes.toBytes(rowkey));
                get.addFamily(Bytes.toBytes(family));
                gets.add(get);
            }
            List<Map<String, String>> returnList = new ArrayList<Map<String, String>>();
            Result[] results = table.get(gets);
            for (Result result : results) {
                //获取列簇,每一个columen的所有版本的值
                Map<String, String> resultMap = noVersionResultToMap(result, family);
                if (!CollectionUtils.isEmpty(resultMap)) {
                    returnList.add(resultMap);
                }
            }
            return returnList;
        } catch (Exception e) {
            logger.info("get value for table: " + tableName + " family: " + family + "failed");
            return null;
        } finally {
            close(table, null);
        }
    }


    /**
     * 根据rowKey找到指定的实体类
     *
     * @param tableName   表名
     * @param family      列簇
     * @param rowKey      rowKey
     * @param clazz       实体类
     * @param maxVersions 最大版本值
     * @param columns     查询的列
     * @param <T>         需要返回的实体类
     * @return 实体类
     */
    public <T> T get(String tableName, String family, String rowKey, Class<T> clazz, Integer maxVersions, String... columns) {
        if (StringUtils.isBlank(tableName) || StringUtils.isBlank(family) || StringUtils.isBlank(rowKey) || clazz == null) {
            throw new RuntimeException("param is null,tableName:" + tableName + ",family:" + family + ",rowKey:" + rowKey + ",class:" + clazz);
        }
        TableName tName = TableName.valueOf(tableName);
        Table table = null;
        try {
            table = connection.getTable(tName);
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addFamily(Bytes.toBytes(family));
            if (maxVersions != null) {
                get.setMaxVersions(maxVersions);
            }
            if (columns != null && columns.length != 0) {
                for (String column : columns) {
                    get.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
                }
            }
            Result result = table.get(get);
            Map resultMap;
            if (maxVersions == null) {
                resultMap = noVersionResultToMap(result, family, true);
            } else {
                resultMap = getMultiResultMap(family, result, true);
            }
            T instance = clazz.newInstance();
            BeanUtils.populate(instance, resultMap);
            return instance;
        } catch (Exception e) {
            logger.error("error:" + e.getMessage() + " get value for table: " + tableName + " family: " + family + "failed");
            //logger.error(e.getMessage(), e);
            throw new RuntimeException(e);
        } finally {
            close(table, null);
        }

    }

    public <T> List<T> get(String tableName, String family, Collection<String> rowKeys, Class<T> clazz, Integer maxVersions, String... columns) {
        if (StringUtils.isBlank(tableName) || StringUtils.isBlank(family) || CollectionUtils.isEmpty(rowKeys) || clazz == null) {
            throw new RuntimeException("param is null,tableName:" + tableName + ",family:" + family + ",rowKey:" + rowKeys + ",class:" + clazz);
        }
        long startTime = System.currentTimeMillis();
        TableName tName = TableName.valueOf(tableName);
        Table table = null;
        try {
            table = connection.getTable(tName);
            List<Get> gets = getGets(family, rowKeys, maxVersions, columns);

            Result[] results = table.get(gets);
            logger.info("HBaseUtils.get " + tableName + " cost:" + (System.currentTimeMillis() - startTime) + " ms,keySize:" + rowKeys.size());
            List<T> resultList = Lists.newArrayList();
            for (Result result : results) {
                if (result.isEmpty()) {
                    continue;
                }
                //获取列簇,每一个columen的所有版本的值
                Map resultMap;
                if (maxVersions == null) {
                    resultMap = noVersionResultToMap(result, family, true);
                } else {
                    resultMap = getMultiResultMap(family, result, true);
                }
                if (MapUtils.isEmpty(resultMap)) {
                    continue;
                }
                T instance = clazz.newInstance();
                BeanUtils.populate(instance, resultMap);
                resultList.add(instance);
            }
            return resultList;
        } catch (Exception e) {
            logger.error("error:" + e.getMessage() + " get value for table: " + tableName + " family: " + family + "failed");
            //logger.error(e.getMessage(), e);
            throw new RuntimeException(e);
        } finally {
            close(table, null);
        }

    }

    private List<Get> getGets(String family, Collection<String> rowKeys, Integer maxVersions, String[] columns) throws IOException {
        List<Get> gets = new ArrayList<Get>();
        for (String rowkey : rowKeys) {
            Get get = new Get(Bytes.toBytes(rowkey));
            get.addFamily(Bytes.toBytes(family));
            if (maxVersions != null) {
                get.setMaxVersions(maxVersions);
            }
            if (columns != null && columns.length != 0) {
                for (String column : columns) {
                    get.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
                }
            }
            gets.add(get);
        }
        return gets;
    }

    /**
     * 根据rowKey找到指定的实体类
     *
     * @param tableName 表名
     * @param family    列簇
     * @param rowKey    rowKey
     * @param clazz     实体类
     * @param columns   查询的列
     * @param <T>       需要返回的实体类
     * @return 实体类
     */
    public <T> T get(String tableName, String family, String rowKey, Class<T> clazz, String... columns) {
        return get(tableName, family, rowKey, clazz, null, columns);
    }

    /**
     * 根据指定的key集合和版本数查询数据
     *
     * @param tableName 表名
     * @param family    列簇
     * @param rowKeys   rowKey集合
     * @return List
     */
    public List<Map<String, String>> get(String tableName, String family, List<String> rowKeys, boolean includeRowKey) {
        if (StringUtils.isBlank(tableName) || rowKeys == null || rowKeys.isEmpty() || StringUtils.isBlank(family)) {
            return null;
        }

        TableName tName = TableName.valueOf(tableName);
        Table table = null;
        try {
            table = connection.getTable(tName);
            List<Get> gets = new ArrayList<Get>();
            for (String rowkey : rowKeys) {
                Get get = new Get(Bytes.toBytes(rowkey));
                get.addFamily(Bytes.toBytes(family));
                gets.add(get);
            }
            List<Map<String, String>> returnList = new ArrayList<Map<String, String>>();
            Result[] results = table.get(gets);
            for (Result result : results) {
                //获取列簇,每一个columen的所有版本的值
                Map<String, String> resultMap = noVersionResultToMap(result, family, includeRowKey);
                if (!CollectionUtils.isEmpty(resultMap)) {
                    returnList.add(resultMap);
                }
            }
            return returnList;
        } catch (Exception e) {
            logger.info("get value for table: " + tableName + " family: " + family + "failed");
            return null;
        } finally {
            close(table, null);
        }
    }

    /**
     * 批量查询数据
     *
     * @param tableName     表名
     * @param family        列簇
     * @param rowKeys       批量查询的rowKey
     * @param maxVersions   查询版本数量
     * @param includeRowKey 是否返回rowKey
     * @param columns       查询的列名
     * @return Map集合
     */
    public List<Map<String, List<String>>> get(String tableName, String family, Collection<String> rowKeys, Integer maxVersions, boolean includeRowKey, String... columns) {
        if (StringUtils.isBlank(tableName) || CollectionUtils.isEmpty(rowKeys) || StringUtils.isBlank(family)) {
            throw new RuntimeException("para tableName,rowKeys,family can't not be null or empty!");
        }

        TableName tName = TableName.valueOf(tableName);
        Table table = null;
        try {
            table = connection.getTable(tName);
            List<Get> gets = getGets(family, rowKeys, maxVersions, columns);

            List<Map<String, List<String>>> returnList = new ArrayList<Map<String, List<String>>>();
            Result[] results = table.get(gets);
            for (Result result : results) {
                if (result.isEmpty()) {
                    continue;
                }
                //获取列簇,每一个columen的所有版本的值
                Map<String, List<String>> resultMap = getMultiResultMap(family, result, includeRowKey);
                if (!CollectionUtils.isEmpty(resultMap)) {
                    returnList.add(resultMap);
                }
            }
            return returnList;
        } catch (Exception e) {
            logger.info("get value for table: " + tableName + " family: " + family + "failed");
            return null;
        } finally {
            close(table, null);
        }
    }

    /**
     * 根据指定的key集合和版本数查询数据
     *
     * @param tableName   表名
     * @param family      列簇
     * @param rowKeys     rowKey集合
     * @param maxVersions 最大版本数
     */
    public List<Map<String, List<String>>> get(String tableName, String family, List<String> rowKeys, Integer maxVersions) {
        if (StringUtils.isBlank(tableName) || rowKeys == null || rowKeys.isEmpty() || StringUtils.isBlank(family)) {
            return null;
        }

        TableName tName = TableName.valueOf(tableName);
        Table table = null;
        try {
            table = connection.getTable(tName);
            List<Get> gets = new ArrayList<Get>();
            for (String rowkey : rowKeys) {
                Get get = new Get(Bytes.toBytes(rowkey));
                get.addFamily(Bytes.toBytes(family));
                get.setMaxVersions(maxVersions);
                gets.add(get);
            }
            List<Map<String, List<String>>> returnList = new ArrayList<Map<String, List<String>>>();
            Result[] results = table.get(gets);
            for (Result result : results) {
                //获取列簇,每一个columen的所有版本的值
                Map<String, List<String>> resultMap = getResultMap(family, result);
                if (!CollectionUtils.isEmpty(resultMap)) {
                    returnList.add(resultMap);
                }
            }
            return returnList;
        } catch (Exception e) {
            logger.info("get value for table: " + tableName + " family: " + family + "failed");
            return null;
        } finally {
            close(table, null);
        }
    }

    /**
     * 保存数据到hbase,如果值为空，则不会保存到hbase中
     *
     * @param data      要保存到hbase的Map
     * @param tableName 表名
     * @param family    列簇名称
     * @param rowKey    rowKey
     * @return Boolean
     */
    public Boolean save(Map<String, String> data, String tableName, String family, String rowKey) {
        return save(data, tableName, family, rowKey, null);
    }

    /**
     * 保存数据到hbase,如果值为空，也会保存到hbase中
     *
     * @param data      要保存到hbase的Map
     * @param tableName 表名
     * @param family    列簇名称
     * @param rowKey    rowKey
     * @return Boolean
     */
    public Boolean save(Map<String, String> data, String tableName, String family, String rowKey, boolean withBlank) {
        return save(data, tableName, family, rowKey, null, withBlank);
    }

    /**
     * 保存数据到hbase
     *
     * @param data      要保存到hbase的Map
     * @param tableName 表名
     * @param family    列簇名称
     * @param rowKey    rowKey
     * @param ttl       数据存活时间，单位为毫秒
     * @return Boolean
     */
    public Boolean save(Map<String, String> data, String tableName, String family, String rowKey, Long ttl) {
        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowKey));
            if (ttl != null) {
                put.setTTL(ttl);
            }
            if (CollectionUtils.isEmpty(data)) {
                return false;
            }
            putColnum(data, family, put);
            table.put(put);
            return true;
        } catch (IOException e) {
            logger.error(e.getMessage(), e);
            return false;
        } finally {
            close(table, null);
        }
    }

    public Boolean save(Map<String, String> data, String tableName, String family, String rowKey, Long ttl, boolean withBlank) {
        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowKey));
            if (ttl != null) {
                put.setTTL(ttl);
            }
            if (CollectionUtils.isEmpty(data)) {
                return false;
            }
            putColnum(data, family, put, withBlank);
            table.put(put);
            return true;
        } catch (IOException e) {
            logger.error(e.getMessage(), e);
            return false;
        } finally {
            close(table, null);
        }
    }

    /**
     * 批量保存数据到hbase
     *
     * @param datas     要保存到hbase的集合
     * @param tableName 表名
     * @param family    列簇名称
     * @param ttl       数据存活时间，单位为毫秒
     * @return Boolean
     */

    public Boolean batchSave(Map<String, Map<String, String>> datas, String tableName, String family, Long ttl) {
        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));
            List<Put> puts = new ArrayList<Put>();
            for (String key : datas.keySet()) {
                Put put = new Put(Bytes.toBytes(key));
                if (ttl != null) {
                    put.setTTL(ttl);
                }
                if (CollectionUtils.isEmpty(datas)) {
                    continue;
                }
                putColnum(datas.get(key), family, put);
                puts.add(put);
            }
            table.put(puts);
            return true;
        } catch (IOException e) {
            logger.error(e.getMessage(), e);
            return false;
        } finally {
            close(table, null);
        }
    }

    /**
     * 批量保存数据到hbase
     *
     * @param datas     要保存到hbase的集合
     * @param tableName 表名
     * @param family    列簇名称
     * @return Boolean
     */

    public Boolean batchSave(Map<String, Map<String, String>> datas, String tableName, String family) {
        return batchSave(datas, tableName, family, null);
    }

    private void putColnum(Map<String, String> data, String family, Put put) {
        byte[] familyBytes = Bytes.toBytes(family);
        Set<Entry<String, String>> entrySet = data.entrySet();
        for (Entry<String, String> entry : entrySet) {
            String key = entry.getKey();
            String value = entry.getValue();
            if (StringUtils.isNotBlank(value)) {
                put.addColumn(familyBytes, Bytes.toBytes(key), Bytes.toBytes(value));
            }
        }
    }

    private void putColnum(Map<String, String> data, String family, Put put, boolean withBlank) {
        byte[] familyBytes = Bytes.toBytes(family);
        Set<Entry<String, String>> entrySet = data.entrySet();
        for (Entry<String, String> entry : entrySet) {
            String key = entry.getKey();
            String value = entry.getValue();
            if (withBlank) {
                put.addColumn(familyBytes, Bytes.toBytes(key), Bytes.toBytes(value));
            } else {
                if (StringUtils.isNotBlank(value)) {
                    put.addColumn(familyBytes, Bytes.toBytes(key), Bytes.toBytes(value));
                }
            }
        }
    }

    /**
     * 保存数据到hbase
     *
     * @param data      要保存到hbase的Map
     * @param tableName 表名
     * @param family    列簇名称
     * @param rowKey    rowKey
     * @return Boolean
     */
    public Boolean saveNotExitst(Map<String, String> data, String tableName, String family, String rowKey) {
        Table table = null;

        try {
            table = connection.getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            if (!table.exists(get)) {
                Put put = new Put(Bytes.toBytes(rowKey));
                if (CollectionUtils.isEmpty(data)) {
                    return false;
                }
                putColnum(data, family, put);
                table.put(put);
                return true;
            } else {
                return false;
            }

        } catch (IOException e) {
            e.printStackTrace();
            return false;
        } finally {
            close(table, null);
        }
    }

    /**
     * 查询表中所有记录和列
     *
     * @param tableName 表名
     * @param family    列簇名称
     * @return List
     */
    public List<Map<String, String>> queryAll(String tableName, String family) {
        return queryAll(tableName, family, new String[0]);
    }

    /**
     * 查询表中指定列的所有记录
     *
     * @param tableName 表名
     * @param family    列簇名称
     * @param columns   指定列，可以是多个列，可以传NULL，传NULL表示查询所有列
     * @return List
     */
    public List<Map<String, String>> queryAll(String tableName, String family, String... columns) {
        if (StringUtils.isBlank(tableName) || StringUtils.isBlank(family)) {
            return null;
        }

        List<Map<String, String>> resultList = new ArrayList<Map<String, String>>();
        TableName tName = TableName.valueOf(tableName);
        Table table = null;
        ResultScanner scanner = null;
        try {
            table = connection.getTable(tName);
            Scan scan = new Scan();
            scan.addFamily(Bytes.toBytes(family));
            addColumns(family, scan, columns);
            scanner = table.getScanner(scan);
            //结果处理放入resultList中
            noVersionResultToList(resultList, scanner, family);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            close(table, scanner);
        }
        return resultList;
    }

    /**
     * 查询表中指定列的所有记录
     *
     * @param tableName 表名
     * @param family    列簇名称
     * @param columns   指定列，可以是多个列，可以传NULL，传NULL表示查询所有列
     * @return List
     */
    public List<Map<String, String>> queryAllAndRow(String tableName, String family, String... columns) {
        if (StringUtils.isBlank(tableName) || StringUtils.isBlank(family)) {
            return null;
        }

        List<Map<String, String>> resultList = new ArrayList<Map<String, String>>();
        TableName tName = TableName.valueOf(tableName);
        Table table = null;
        ResultScanner scanner = null;
        try {
            table = connection.getTable(tName);
            Scan scan = new Scan();
            scan.addFamily(Bytes.toBytes(family));
            addColumns(family, scan, columns);
            scanner = table.getScanner(scan);
            //结果处理放入resultList中
            noVersionResultToListAndRow(resultList, scanner, family);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            close(table, scanner);
        }
        return resultList;
    }

    /**
     * 根据指定范围查询表
     *
     * @param tableName
     * @param start_rowkey
     * @param stop_rowkey
     * @return
     * @throws IOException
     */
    public List<Map<String, String>> scanByRange(String tableName, String start_rowkey,
                                                 String stop_rowkey, String family) throws IOException {
        if (StringUtils.isBlank(tableName)) {
            return null;
        }
        List<Map<String, String>> resultList = new ArrayList<Map<String, String>>();
        TableName tName = TableName.valueOf(tableName);
        Table table = null;
        ResultScanner scanner = null;
        try {
            table = connection.getTable(tName);
            Scan scan = new Scan();
            scan.addFamily(Bytes.toBytes(family));
            scan.setStartRow(Bytes.toBytes(start_rowkey));
            scan.setStopRow(Bytes.toBytes(stop_rowkey));
            scanner = table.getScanner(scan);
            //结果处理放入resultList中
            filterByCreateTime(resultList, scanner, family);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            close(table, scanner);
        }

        return resultList;
    }

    private void addColumns(String family, Scan scan, String[] columns) {
        if (columns != null && columns.length > 0) {
            for (String column : columns) {
                if (StringUtils.isNotBlank(column)) {
                    scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
                }
            }
        }
    }

    /**
     * 将scanner结果转为map，并放入List中
     *
     * @param resultList 存结果的集合
     * @param scanner    scanner
     * @param family     列簇
     */
    private void noVersionResultToList(List<Map<String, String>> resultList, ResultScanner scanner, String family) {
        for (Result result : scanner) {
            Map<String, String> resultMap = noVersionResultToMap(result, family);
            if (result != null && result.size() > 0) {
                resultList.add(resultMap);

            }
        }
    }

    private void filterByCreateTime(List<Map<String, String>> resultList, ResultScanner scanner, String family) {
        for (Result result : scanner) {
            Map<String, String> resultMap = noVersionResultToMap(result, family);
            if (result != null && result.size() > 0) {
                if (resultMap.get("insert_time") != null || resultMap.get("create_time") != null) {
                    resultList.add(resultMap);
                } else if (resultMap.get("updateTime") != null || resultMap.get("updateTime") != null) {
                    resultList.add(resultMap);
                }
            }
        }
    }

    /**
     * 将scanner结果转为map，并放入List中
     *
     * @param resultList 存结果的集合
     * @param scanner    scanner
     * @param family     列簇
     */
    private void noVersionResultToListAndRow(List<Map<String, String>> resultList, ResultScanner scanner, String family) {
        for (Result result : scanner) {
            Map<String, String> resultMap = noVersionResultToMap(result, family);
            resultMap.put("rowkey", new String(result.getRow()));
            if (result != null && result.size() > 0) {
                resultList.add(resultMap);
            }
        }
    }

    /**
     * 将scanner结果转为map，并放入List中
     *
     * @param result scanner结果
     * @param family 列簇
     * @return Map
     */
    private Map<String, String> noVersionResultToMap(Result result, String family) {

        NavigableMap<byte[], NavigableMap<byte[], byte[]>> map = result.getNoVersionMap();
        if (CollectionUtils.isEmpty(map)) {
            return null;
        }
        NavigableMap<byte[], byte[]> familyMap = map.get(Bytes.toBytes(family));
        Set<Entry<byte[], byte[]>> entrySet = familyMap.entrySet();
        Map<String, String> resultMap = new HashMap<String, String>();

        for (Entry<byte[], byte[]> entry : entrySet) {
            byte[] key = entry.getKey();
            byte[] value = entry.getValue();
            resultMap.put(Bytes.toString(key), Bytes.toString(value));
        }
        return resultMap;
    }

    /**
     * 根据指定KEY查询最新版本的记录
     *
     * @param tableName 表名
     * @param family    列簇
     * @param rowKey    rowKey
     * @return List
     */
    public Map<String, String> queryNoVersion(String tableName, String family, String rowKey) {
        if (StringUtils.isBlank(rowKey) || StringUtils.isBlank(tableName) || StringUtils.isBlank(family)) {
            return null;
        }
        TableName tName = TableName.valueOf(tableName);
        Table table = null;
        try {
            table = connection.getTable(tName);
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addFamily(Bytes.toBytes(family));
            //结果处理放入resultList中
            return noVersionResultToMap(table.get(get), family);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            close(table);
        }
        return null;
    }


    

    /*@SuppressWarnings("unused")
    private List<Map<String, String>> queryAndCache(String key,String tableName, String family) {
        List<Map<String, String>> list = queryAll(tableName, family);
        if (list != null && !list.isEmpty()) {
            jedisWriterTemplate.setArray(key, list, 24 * 60 * 60 * 2);
        }
        return list;
    }*/

    private void close(Table table, ResultScanner scanner) {
        try {
            if (null != scanner) {
                scanner.close();
            }
            if (null != table) {
                table.close();
            }
        } catch (Exception e2) {
            e2.printStackTrace();
            logger.info("close connection failed");
        }
    }

    private void close(Table table) {
        close(table, null);
    }

    /**
     * 查询push_history_device表中30天的rowKeys
     *
     * @param tableName 表名
     * @param family    列簇
     * @param rowKeys   rowKey集合
     * @return
     */
    public Set<String> query(String tableName, String family, List<String> rowKeys) {

        Set<String> cityCodes = new HashSet<String>();

        Table table = null;
        try {
            TableName tName = TableName.valueOf(tableName);
            table = connection.getTable(tName);
            for (String rowKey : rowKeys) {
                Get get = new Get(Bytes.toBytes(rowKey));
                if (table.exists(get)) {
                    Result result = table.get(get);
                    byte[] cityCode = result.getValue(Bytes.toBytes(family), Bytes.toBytes("cityId"));
                    if (null != cityCode) {
                        cityCodes.add(new String(cityCode));
                    }
                }
            }
            return cityCodes;
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        } finally {
            if (null != table) {
                try {
                    table.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @SuppressWarnings("resource")
    public long rowCount(String tableName, String family, FilterList filterList) {

        AggregationClient ac = new AggregationClient(configuration);
        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(family));

        scan.setFilter(filterList);
        long rowCount = 0;
        try {
            rowCount = ac.rowCount(TableName.valueOf(tableName), new LongColumnInterpreter(), scan);
        } catch (Throwable e) {
            e.printStackTrace();
            logger.info("rowCount for table:" + tableName + "failed!");
        }
        return rowCount;
    }


}
